package com.xiaozhicha.mq.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.junit.Before;
import org.junit.Test;

import java.time.Duration;
import java.util.Arrays;
import java.util.Iterator;
import java.util.Properties;

public class ConsumeTest {

    private Properties properties;

    @Before
    public void init(){
        properties = new Properties();
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("group.id", "testConsumer");
        properties.put("enable.auto.commit", "true");
        properties.put("auto.commit.interval.ms", "1000");
        properties.put("session.timeout.ms", "30000");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    }

    @Test
    public void consume(){
        System.out.println("begin consumer");
        connectionKafka();
        System.out.println("finish consumer");
    }

    public void connectionKafka(){
        KafkaConsumer<String,String> consumer = new KafkaConsumer(properties);
        consumer.subscribe(Arrays.asList("my-topic","test-topic","test0811"));
        while (true){
            ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(1));
            Iterator<ConsumerRecord<String,String>> it = records.iterator();
            while (it.hasNext()){
                System.out.println("收到消息" + it.next().value());
            }
        }
    }
}
